package com.acegear.horizon.domain.events.processers;


import com.acegear.horizon.domain.events.BaseEvent;
import com.acegear.horizon.vendor.ons.ONSConsumer;
import com.acegear.horizon.vendor.ons.ONSPublisher;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.function.Consumer;

import reactor.core.Disposable;
import reactor.core.publisher.BlockingSink;
import reactor.core.publisher.EmitterProcessor;

/**
 * Created by mercury on 2016/12/14.
 */
@Component
public class EventProcessor<T extends BaseEvent> {

    private EmitterProcessor<T> emitterProcessor = EmitterProcessor.create();
    private BlockingSink<T> sink = emitterProcessor.connectSink();


    @Autowired
    private ONSPublisher publisher;

    @Autowired
    private ONSConsumer consumer;

    public EventProcessor() {
    }

    public Disposable subscribeOnce(Class<T> tClass, Consumer<T> consumer) {
        return emitterProcessor.filter(tClass::isInstance).publishNext().subscribe(consumer);
    }

    public Disposable subscribeOnce(Class<T> tClass, Consumer<T> consumer, Consumer<Throwable> error) {
        return emitterProcessor.filter(tClass::isInstance).publishNext().subscribe(consumer, error);
    }

    public Disposable subscribeOnce(Class<T> tClass, Consumer<T> consumer,
                                    Consumer<Throwable> errorConsumer,
                                    Runnable completeConsumer) {
        return emitterProcessor.filter(tClass::isInstance).publishNext().subscribe(consumer, errorConsumer, completeConsumer);
    }

    public Disposable subscribe(Class<T> tClass, Consumer<T> consumer) {
        return emitterProcessor.filter(tClass::isInstance).subscribe(consumer);
    }

    public Disposable subscribe(Class<T> tClass, Consumer<T> consumer, Consumer<Throwable> error) {
        return emitterProcessor.filter(tClass::isInstance).subscribe(consumer, error);
    }

    public Disposable subscribe(Class<T> tClass, Consumer<T> consumer,
                                Consumer<Throwable> errorConsumer,
                                Runnable completeConsumer) {
        return emitterProcessor.filter(tClass::isInstance).subscribe(consumer, errorConsumer, completeConsumer);
    }

    public <E extends BaseEvent> void consumeONS(Class<E> tClass, Consumer<E> con) {
        consumer.consume(tClass,con);
    }

    public void publish(T t) {
        publisher.sendMessage(t);
    }
}
